<?php
/**
 * Simq 队列服务客户端类
 */
class Simq
{
    /**
     * 配置
     */
    protected $_server = array(
        'host' => '127.0.0.1',
        'port' => 27777,
        'auth' => 'abc123',
    );

    /**
     * 构造
     */
    public function __construct($config = null)
    {
        if (is_array($config)) {
            foreach ($config as $key => $value) {
                if (isset($this->_server[$key])) {
                    $this->_server[$key] = $value;
                }
            }
        }
    }

    // 推送消息
    public function put($queue, $msg)
    {
        // 生成uri
        $uri = sprintf('/%s', $queue);

        // 推送到simq
        $r = $this->_http('PUT', $uri, $msg);

        // 返回结果
        return ($r !== false) && ($r['Count'] > 0);
    }

    // 获取消息
    public function get($queue)
    {
        // 生成uri
        $uri = sprintf('/%s', $queue);

        // 推送到simq
        $r = $this->_http('GET', $uri);

        // 返回结果
        if ($r === false) {
            return null;
        }

        return $r['Body'];
    }

    // 获取队列长度
    public function len($queue)
    {
        // 生成uri
        $uri = sprintf('/%s', $queue);

        // 推送到simq
        $r = $this->_http('HEAD', $uri);

        // 返回结果
        if ($r === false) {
            return 0;
        }

        return $r['Count'];
    }

    // 订阅
    public function sub($queue, $url)
    {
        // 生成uri
        $uri = sprintf('/%s', $queue);

        // 推送到simq
        $r = $this->_http('SUB', $uri, $url);

        // 返回结果
        return ($r !== false) && ($r['Count'] > 0);
    }

    // 退订
    public function unsub($queue, $url)
    {
        // 生成uri
        $uri = sprintf('/%s', $queue);

        // 推送到simq
        $r = $this->_http('UNSUB', $uri, $url);

        // 返回结果
        return ($r !== false) && ($r['Count'] > 0);
    }

    // 连接 simq 服务
    protected function _http($method, $uri, $data = null, $timeout = 30)
    {
        $fsock = fsockopen($this->_server['host'], $this->_server['port'], $errno, $errstr, $timeout);
        if (!$fsock) {
            return false;
        }

        $method = strtoupper($method);

        $in = $method . ' ' . $uri . " HTTP/1.1\r\n";
        $in = $in . 'Auth: ' . $this->_server['auth'] . "\r\n";

        if ($data) {
            $in .= "Content-Length: " . strlen($data) . "\r\n\r\n";
            $in .= $data;
        } else {
            $in .= "\r\n";
        }

        if (!@fwrite($fsock, $in, strlen($in))) {
            fclose($fsock);
            return false;
        }

        $out = null;
        while ($buff = @fgets($fsock, 2048)) {
            $out .= $buff;
        }
        fclose($fsock);

        $arr = array();

        $pos1      = strpos($out, "\r\n");
        $firstline = substr($out, 0, $pos1);
        if (preg_match("/^HTTP\/\d\.\d\s(\d+)\s.*$/", $firstline, $matches)) {
            $arr['Status'] = $matches[1];
        }

        $pos1 += 2;
        $pos2 = strpos($out, "\r\n\r\n", $pos1);

        $headers     = substr($out, $pos1, $pos2 - $pos1);
        $arr['Body'] = substr($out, $pos2 + 4);

        $headers = trim($headers);
        if (!empty($headers)) {
            $lines = explode("\r\n", $headers);
            foreach ($lines as $line) {
                if (empty($line)) {
                    continue;
                }

                list($key, $value) = explode(':', $line);
                $arr[$key]         = trim($value);
            }
        }

        return $arr;
    }
}
